package ar.edu.itba.pod.legajo48013.balance;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import ar.edu.itba.balance.api.AgentsBalancer;
import ar.edu.itba.balance.api.AgentsTransfer;
import ar.edu.itba.balance.api.NodeAgent;
import ar.edu.itba.balance.api.NotCoordinatorException;
import ar.edu.itba.node.Node;
import ar.edu.itba.node.NodeInformation;
import ar.edu.itba.pod.legajo48013.NodeAgentsInformation;
import ar.edu.itba.pod.legajo48013.Utils;
import ar.edu.itba.pod.legajo48013.event.DisconnectThread;
import ar.edu.itba.pod.legajo48013.host.Host;
import ar.edu.itba.pod.legajo48013.rmi.LookupUtil;

public class MyAgentsBalancer extends UnicastRemoteObject implements AgentsBalancer {

	private static final long serialVersionUID = -890127339200476021L;
	
	private final Host host;
	private AtomicBoolean threwElection = new AtomicBoolean();
	private int MILLISECONDS_TO_PASS = 3000;
	private NodeInformation coordinator;
	
	private List<NodeAgentsInformation> agentsInformation;	
	private ConcurrentMap<NodeInformation, Long> elections = new ConcurrentHashMap<NodeInformation, Long>();

	private CountDownLatch latch = new CountDownLatch(1);
	
	public MyAgentsBalancer(final Host host) throws RemoteException {
		super();
		this.host = host;
	}
	
	public void startElection(NodeInformation node, long timestamp){
		Set<NodeInformation> copyNodesConnected = Utils.getCopyOfConnectedNodes(host);
		threwElection.set(true);
		coordinator = null;
		new Thread(new SendElectionRunnable(copyNodesConnected, timestamp)).start();
	}
	
	@Override
	public void bullyElection(NodeInformation node, long timestamp) throws RemoteException {
		if(!haveElection(node, timestamp) && !node.equals(host.getNode())){
			coordinator = null;
			elections.put(node, timestamp);
			if(threwElection.compareAndSet(false, true)){
				if( node.id().compareTo(host.getNode().id()) < 0 ) {
					AgentsBalancer balancer = (AgentsBalancer) LookupUtil.lookup(host, node, Node.AGENTS_BALANCER);
					if(balancer != null){
						System.out.println("sending ok from " + host.getNode() + " to " + node);
						balancer.bullyOk(host.getNode());
					}
				}
				Set<NodeInformation> copyNodesConnected = Utils.getCopyOfConnectedNodes(host);
				new Thread(new SendElectionRunnable(copyNodesConnected, System.currentTimeMillis())).start();
			}
		}
	}

	@Override
	public void bullyOk(NodeInformation node) throws RemoteException {
		System.out.println("getting ok from " + node);
		threwElection.set(false);
		latch.countDown();
	}

	@Override
	public void bullyCoordinator(NodeInformation node, long timestamp) throws RemoteException {
		System.out.println("in node " + host.getNode() + ". I have " + elections);
		System.out.println(elections.size());
		System.out.println("receiving coordinator " + node + " time: " + timestamp);
		if(!haveElection(node, timestamp)){
			this.coordinator=node;
			System.out.println("setting coordinator" + node);
			elections.put(node, timestamp);
			Set<NodeInformation> copyOfNodesConnected = Utils.getCopyOfConnectedNodes(host);
//			if(this.coordinator.equals(host.getNode())){
//				this.agentsInformation = new ArrayList<NodeAgentsInformation>();
//				getNodeInformations(copyOfNodesConnected);
//			}else{
//				this.agentsInformation = null;
//			}
//			elections.put(host.getNode(), System.currentTimeMillis());
			new Thread(new SendCoordinatorRunnable(copyOfNodesConnected, timestamp)).start();
		}

	}

	private boolean haveElection(NodeInformation node, long timestamp){
		Long time = elections.get(node);
		if(time!=null)
			return time.equals(timestamp);
		return false;
	}
	
	private void getNodeInformations(final Set<NodeInformation> nodesConnected) throws RemoteException{
		new Thread(){
			public void run() {
				for(NodeInformation nodeConnected: nodesConnected){
					AgentsTransfer transfer = (AgentsTransfer) LookupUtil.lookup(host, nodeConnected, Node.AGENTS_TRANSFER);
					if(transfer != null){
						try {
							int amount = transfer.getNumberOfAgents();
							agentsInformation.add(new NodeAgentsInformation(nodeConnected, amount));
						} catch (RemoteException e) {
							new DisconnectThread(host, nodeConnected).start();
						}
					}
				}
				balance(agentsInformation, null);
			}
		}.start();
	}
	
	private class SendElectionRunnable extends BullyRunnable{
		public SendElectionRunnable(Set<NodeInformation> nodesConnected, long timestamp) {
			super(nodesConnected, timestamp);
		}
		@Override
		public void send(AgentsBalancer balancer) throws RemoteException {
			System.out.println("sending election from " + host.getNode());
			balancer.bullyElection(host.getNode(), super.timestamp);
		}

		@Override
		public void waitTime() {
			try{
				if(latch.await(MILLISECONDS_TO_PASS, TimeUnit.MILLISECONDS)){
					System.out.println("received OK");
					return;
				}else{
					System.out.println("nadie me mando ok");
					try{
						bullyCoordinator(host.getNode(), System.currentTimeMillis());
					} catch (RemoteException e) {
						e.printStackTrace();
						//calling my own bully coordinator. shouldn't throw exception
					}
				}
			} catch (InterruptedException e) { //interrupted due to virtual machine or something else
				e.printStackTrace();
			}
		}
	}
	
	private class SendCoordinatorRunnable extends BullyRunnable{
		public SendCoordinatorRunnable(Set<NodeInformation> nodesConnected, long timestamp) {
			super(nodesConnected, timestamp);
		}
		@Override
		public void send(AgentsBalancer balancer) throws RemoteException {
			System.out.println("sending coordinator" + coordinator + "");
			balancer.bullyCoordinator(coordinator, super.timestamp);
		}
		@Override
		public void waitTime() {
			return;
		}
	}

	private abstract class BullyRunnable implements Runnable{
		private Set<NodeInformation> nodesConnected;
		private long timestamp;
		public BullyRunnable(Set<NodeInformation> nodesConnected, long timestamp) {
			this.nodesConnected = nodesConnected;
			this.timestamp = timestamp;
		}
		
		public abstract void send(AgentsBalancer balancer) throws RemoteException;
		
		public abstract void waitTime();
		
		@Override
		public void run(){
			for(NodeInformation nodeConnected: nodesConnected){
				AgentsBalancer balancer = (AgentsBalancer) LookupUtil.lookup(host, nodeConnected, Node.AGENTS_BALANCER);
				if(balancer != null){
					try {
						System.out.println("nodeConnected " + nodeConnected);
						send(balancer);
					} catch (RemoteException e) {
						new DisconnectThread(host, nodeConnected).start();
					}
				}
			}
			waitTime();
		}			
	}

	private void checkIfImCoordinator() throws NotCoordinatorException{
		if(!host.getNode().equals(coordinator))
			throw new NotCoordinatorException(coordinator);
	}

	private void balance(List<NodeAgentsInformation> agentsInformation, List<NodeAgent> removedAgents){
		int am = 0;
		for(NodeAgentsInformation nodeAgentInfo: agentsInformation){
			am += nodeAgentInfo.getAmount();
		}
		float avgReal = am / agentsInformation.size();
		int avgInt = (int)avgReal; //minimum amount of agents each node should have
		int rest = (int)(avgReal-avgInt)*agentsInformation.size(); //amount of nodes that have 1 more agent than the rest
		List<NodeAgent> nodeAgents = new ArrayList<NodeAgent>();
		if(removedAgents!=null){
			nodeAgents = removedAgents;
		}
	
		Comparator<NodeAgentsInformation> comp = Collections.reverseOrder();
		Collections.sort(agentsInformation, comp);
		if(checkIfUnmbalanced(agentsInformation)){
			for(NodeAgentsInformation nodeAgentInfo: agentsInformation){
				try{
					int dif = nodeAgentInfo.getAmount() - avgInt - (rest>0?1:0);
					NodeInformation nodeInfo = nodeAgentInfo.getNode();
					AgentsTransfer transfer = (AgentsTransfer)LookupUtil.lookup(host, nodeInfo, Node.AGENTS_TRANSFER);
					if(transfer!=null){
						if(dif<0){
							nodeAgents.addAll(transfer.stopAndGet(dif*-1));
						}else{
							transfer.runAgentsOnNode(nodeAgents.subList(0, dif));
						}
						rest--;
					}
				}catch (RemoteException e) {
					new DisconnectThread(host, nodeAgentInfo.getNode()).start();
				}
			}
		}
		
	}
	
	private boolean checkIfUnmbalanced(List<NodeAgentsInformation> agentsInformation){
		return agentsInformation.get(0).getAmount() - agentsInformation.get(agentsInformation.size()-1).getAmount() > 1;
	}

	
	@Override
	public void shutdown(List<NodeAgent> agents) throws RemoteException, NotCoordinatorException {
		checkIfImCoordinator();
		
		//TODO
	}

	@Override
	public void addAgentToCluster(NodeAgent agent) throws RemoteException, NotCoordinatorException {
		checkIfImCoordinator();
		Collections.sort(agentsInformation);
		NodeAgentsInformation nodeToAddAgent = agentsInformation.get(0);
		nodeToAddAgent.addAgent();
		AgentsTransfer transfer = (AgentsTransfer)LookupUtil.lookup(host, nodeToAddAgent.getNode(), Node.AGENTS_TRANSFER);
		if(transfer!=null){
			List<NodeAgent> agents = new ArrayList<NodeAgent>();
			agents.add(agent);
			transfer.runAgentsOnNode(agents);
		}
	}

}
